﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using Sinacor.OMM.Router.ProcessEntity;
using Sinacor.OMM.ConnectionManager;
using Sinacor.OMM.HealthMonitor.Collector;
using Sinacor.OMM.HealthMonitor.Collector.SharedHealthDataMemoryStorage.Router;
using System.Diagnostics;
using Sinacor.OMM.ConnectionManager.MSMQ;
using System.Collections.Specialized;
using Sinacor.OMM.Util;
using PSE.Framework.Threading;

namespace Sinacor.OMM.Router.ProcessLogic
{
    public class MessageProcessor : IDisposable
    {
        private Util.TraceWriter _traceWriter;
        private Util.TraceWriter _traceWriterOtimizadas;
        /// <summary>
        /// 
        /// </summary>
        private const string ROUTER_MESSAGE_PROCESSOR_QUEUE_ID = "RouterMessageProcessor";

        /// <summary>
        /// 
        /// </summary>
        private const int MAX_MESSAGES_PER_BLOCK = 1000;

        /// <summary>
        /// Controls the accessibility to the _messages object.
        /// </summary>
        private object _lock = new object();

        /// <summary>
        /// 
        /// </summary>
        private bool _shouldStopProcessing;

        /// <summary>
        /// 
        /// </summary>
        private Sinacor.OMM.Router.ResourceAccess.Order _orderResourceAccess;

        /// <summary>
        /// 
        /// </summary>
        private List<HftProvider> _providersList;

        /// <summary>
        /// 
        /// </summary>
        private ConnectionManager.MSMQ.MsmqSender _systemMessageProvider;

        /// <summary>
        /// 
        /// </summary>
        public event RouterEventHandler OnError;

        /// <summary>
        /// 
        /// </summary>
        public event RouterEventHandler OnRecovery;        

        /// <summary>
        /// 
        /// </summary>
        private ConnectionManager.MSMQ.MSMQHelper _msmqHelper;

        /// <summary>
        /// 
        /// </summary>
        private readonly char[] _cancellingStatus;

        /// <summary>
        /// 
        /// </summary>
        private string _lastMegaBolsaMessageId;

        /// <summary>
        /// 
        /// </summary>
        private string _lastCblcMessageId;

        /// <summary>
        /// 
        /// </summary>
        private ConnectionManager.InternalQueue.InternalQueueManager<MessageEntity.BaseMessage> _internalQueueManager;

        /// <summary>
        /// 
        /// </summary>
        private DictionaryOrderManager _dictionaryOrderManager;

        /// <summary>
        /// 
        /// </summary>
        private bool _getRepositorySucceeded;

        /// <summary>
        /// 
        /// </summary>
        private RouterShdms _routerShdms;

        /// <summary>
        /// 
        /// </summary>
        private long _routedMessagesToNormalFlow = 0;

        /// <summary>
        /// 
        /// </summary>
        private long _optimizedMessages = 0;

        /// <summary>
        /// 
        /// </summary>
        private long _hftMessages = 0;

        /// <summary>
        /// 
        /// </summary>
        private int _totalRoutedMessagesPerInterval = 0;

        /// <summary>
        /// 
        /// </summary>
        private int _totalRoutedMessages = 0;

        /// <summary>
        /// 
        /// </summary>
        private DateTime _lastHealthStatusChecked = new DateTime(1, 1, 1);

        /// <summary>
        /// 
        /// </summary>
        private DateTime _lastHealthDBStatusChecked = new DateTime(1, 1, 1);

        /// <summary>
        /// 
        /// </summary>
        private int _bulkInsertSize = 0;

        /// <summary>
        /// 
        /// </summary>
        private int _bulkInsertTimes = 0;

        /// <summary>
        /// 
        /// </summary>
        private int _insertOperationTime = 0;

        /// <summary>
        /// 
        /// </summary>
        private ConnectionManager.DataBase.DataBaseCommunicator _dataBaseCommunicator;

        /// <summary>
        /// 
        /// </summary>
        private Dictionary<string, bool> _queueStatusChanged;

        private ThreadStart _databaseCountHandler;

        private Thread _databaseCountThread;

        private bool _shouldStopCheckingDatabaseQueue = false;

        private bool _disposing = false;

        private string _lastProcessedMessageId = string.Empty;

        public string LastProcessedMessageId
        {
            get
            {
                return GetLastProcessedMessageId();
            }
        }

        private DumpClient.Dump _dumpClient;

        private bool _shouldStopCollecting;

        public MessageProcessor()
        {
            // Router statistics
            _routerShdms = RouterShdms.GetSharedStorage();
            _routerShdms.RouterDataInfo.MessageRouterInfo.LimitInternalQueuedMessages_02_Quantity = 200000;
            _routerShdms.RouterDataInfo.DataBaseProcessInfo.LimitDataBaseInsertAverageTime = 1;
            _routerShdms.RouterDataInfo.StopedBySomeQueueExceedsLimit = false;

            _routerShdms.RouterDataInfo.DataBaseProcessInfo.LimitMessagesDataBaseQueueQuantity = 24000000;

            // _traceWriter
            _traceWriter = Tracer.GetInstance("Roteador", "MessageProcessor.log");
            _traceWriterOtimizadas = Tracer.GetInstance("Roteador", "MessageProcessorOtimizadas ");
            // _dumpClient
            _dumpClient = DumpClient.Dump.GetInstance();

            // resource access
            ConnectionManager.DataBase.OracleDatabase oracleDatabase = new Sinacor.OMM.ConnectionManager.DataBase.OracleDatabase();
            oracleDatabase.CompanyId = RouterConfigurationManager.Authentication.CompanyId;
            _dataBaseCommunicator = new Sinacor.OMM.ConnectionManager.DataBase.DataBaseCommunicator(oracleDatabase, "OMMRouterService");
            _orderResourceAccess = new Sinacor.OMM.Router.ResourceAccess.Order(_dataBaseCommunicator);

            _msmqHelper = new Sinacor.OMM.ConnectionManager.MSMQ.MSMQHelper();
            _dictionaryOrderManager = DictionaryOrderManager.GetInstance();
            // _internalQueueManager
            _internalQueueManager = ConnectionManager.InternalQueue.InternalQueueManager<MessageEntity.BaseMessage>.GetInstance();
            _internalQueueManager.SetMaxItensInBlockedEvent(ROUTER_MESSAGE_PROCESSOR_QUEUE_ID, MAX_MESSAGES_PER_BLOCK);
            _internalQueueManager.SetMessageArrival(ROUTER_MESSAGE_PROCESSOR_QUEUE_ID, DequeueNormalFlowMessage);

            _queueStatusChanged = new Dictionary<string, bool>();
            _cancellingStatus = GetCancellingStatus();
            _systemMessageProvider = _msmqHelper.CreateRepository(RouterConfigurationManager.RouterConfiguration.SystemMessageRepository);

            LoadProviders();

            _databaseCountHandler = new ThreadStart(CheckDatabaseQueueCount);
            _databaseCountThread = new Thread(_databaseCountHandler);
            _databaseCountThread.Start();

            // collectorThread
            Thread collectorThread = new Thread(new ThreadStart(Collect));
            collectorThread.Start();

            // _lastProcessedMessageId
            _lastProcessedMessageId = GetLastProcessedMessageId();
            // _lastMegaBolsaMessageId
            SetLastMegaBolsaMessageId();
            // _lastCblcMessageId
            SetLastCblcMessageId();
        }

        private void CheckDatabaseQueueCount()
        {
            while (!_shouldStopCheckingDatabaseQueue)
            {
                try
                {
                    _routerShdms.RouterDataInfo.DataBaseProcessInfo.MessagesDataBaseQueueQuantity = _orderResourceAccess.GetDatabaseQueueCount(
                        RouterConfigurationManager.OrderBovespaSystemInformation.SystemDate);
                }
                catch (Exception ex)
                {
                    _traceWriter.Trace(string.Concat("Error when trying to check database queue count. ", ex.Message, ex.StackTrace));
                    HandleError(1, string.Concat("Error when trying to check database queue count. ", ex.Message, ex.StackTrace));
                }
                Sleeper.Sleep(15000, () => !_shouldStopCheckingDatabaseQueue);
            }
        }

        private void Collect()
        {
            while (!_shouldStopCollecting && !_disposing)
            {
                // Message Router                
                _routerShdms.RouterDataInfo.MessageRouterInfo.RoutedMessagesForSecond = CalculateRoutedMessagesPerSecond();
                _routerShdms.RouterDataInfo.MessageRouterInfo.DefaultFlowRoutedMessagesQuantity = _routedMessagesToNormalFlow;
                _routerShdms.RouterDataInfo.MessageRouterInfo.OptimizedRoutedMessagesQuantity = _optimizedMessages;
                _routerShdms.RouterDataInfo.MessageRouterInfo.HftRoutedMessagesQuantity = _hftMessages;

                lock (_dictionaryOrderManager)
                {
                    _routerShdms.RouterDataInfo.MessageRouterInfo.MemoryControlledOrdersQuantity = _dictionaryOrderManager.Count;
                }

                _routerShdms.RouterDataInfo.MessageRouterInfo.InternalQueuedMessages_02_Quantity = _internalQueueManager.Count(ROUTER_MESSAGE_PROCESSOR_QUEUE_ID);

                // Database
                _routerShdms.RouterDataInfo.DataBaseProcessInfo.BlockProcessingAverageQuantity = CalculateBulkInsertSizePerSecond();
                _routerShdms.RouterDataInfo.DataBaseProcessInfo.DataBaseInsertAverageTime = CalculateInsertAverageTime();

                if (_dataBaseCommunicator != null)
                {
                    _routerShdms.RouterDataInfo.DataBaseProcessInfo.DataBaseSucessfulConnectionRetries = _dataBaseCommunicator.ConnectionTimes;
                    _routerShdms.RouterDataInfo.DataBaseProcessInfo.DataBaseUnsucessfulConnectionRetries = _dataBaseCommunicator.ConnectionWithError;
                }
                else
                {
                    _routerShdms.RouterDataInfo.DataBaseProcessInfo.DataBaseSucessfulConnectionRetries = 0;
                    _routerShdms.RouterDataInfo.DataBaseProcessInfo.DataBaseUnsucessfulConnectionRetries = 0;
                }

                Sleeper.Sleep(1000, () => !_shouldStopCollecting);
            }
        }

        private int CalculateInsertAverageTime()
        {
            int insertAverageTime = 0;

            if (_bulkInsertTimes > 0)
                insertAverageTime = _insertOperationTime / _bulkInsertTimes;
            else
                insertAverageTime = 0;

            return insertAverageTime;
        }

        private int CalculateBulkInsertSizePerSecond()
        {
            int bulkInsertSizePerSecond;

            if (_bulkInsertSize > _bulkInsertTimes)
                bulkInsertSizePerSecond = _bulkInsertSize / _bulkInsertTimes;
            else
                bulkInsertSizePerSecond = _bulkInsertSize;

            return bulkInsertSizePerSecond;
        }

        private int CalculateRoutedMessagesPerSecond()
        {
            int seconds;
            int routedMessagesPerSecond;

            if (_lastHealthStatusChecked.Year == 1)
                _lastHealthStatusChecked = DateTime.Now;

            // Calcula o número de segundos entre a última coleta e a coleta atual
            seconds = DateTime.Now.Subtract(_lastHealthStatusChecked).Seconds;

            if (seconds == 0)
                seconds = 1;

            if (_totalRoutedMessagesPerInterval >= seconds)
                routedMessagesPerSecond = _totalRoutedMessagesPerInterval / seconds;
            else
                routedMessagesPerSecond = _totalRoutedMessagesPerInterval;

            // Zera o contador que será usado na próxima coleta
            // Esse contador é incrementado dentro do Run()
            // a cada mensagem recebida
            _totalRoutedMessagesPerInterval = 0;

            // Define o horário dessa verificação, para ser usada na próxima chamada da função e calcular os segundos entre uma chamada e outra
            _lastHealthStatusChecked = DateTime.Now;

            return routedMessagesPerSecond;
        }

        /// <summary>
        /// 
        /// </summary>
        /// <returns></returns>
        private char[] GetCancellingStatus()
        {
            return new char[] { 'E', 'S', 'O', 'A', 'R', 'C' };
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="message"></param>
        public void Process(MessageEntity.BaseMessage message)
        {
            MessageEntity.STM.BaseMessage msg = (MessageEntity.STM.BaseMessage)message;
            DateTime messageIdDate = new DateTime(int.Parse(message.ID.Substring(0, 4)), int.Parse(message.ID.Substring(4, 2)), int.Parse(message.ID.Substring(6, 2)));

            if (messageIdDate.CompareTo(RouterConfigurationManager.OrderBovespaSystemInformation.SystemDate) == 0)
            {
                if (msg is MessageEntity.STM.GenericMessage)
                {
                    _traceWriter.Trace(string.Concat("The message [", msg.SequentialID, "|", msg.ID, "] is generic. [", msg.Original, "]"), TraceType.Optional);
                    SendMessageToNormalFlow(msg);
                }
                else
                {
                    if (msg.Source == Sinacor.OMM.MessageEntity.Enum.MessageSource.MegaBolsa)
                    {
                        if (!(msg is MessageEntity.STM.Message0100))
                        {
                            if (IsProcessable(message))
                            {
                                if (IsHFT(message))
                                {
                                    if (msg is MessageEntity.STM.Message0103)
                                        ProcessMessage0103(msg);

                                    if (msg is MessageEntity.STM.Message0105)
                                        ProcessMessage0105(msg);

                                    if (msg is MessageEntity.STM.Message0138)
                                        ProcessMessage0138(msg);

                                    if (msg is MessageEntity.STM.Message0172)
                                        ProcessMessage0172(msg);
                                }
                                else
                                {
                                    _traceWriter.Trace(string.Concat("The message [", msg.SequentialID, "|", message.ID, "] isn't HFT. [", msg.Original, "]"), TraceType.Optional);
                                    SendMessageToNormalFlow(message);
                                }
                            }
                            else
                            {
                                _traceWriter.Trace(string.Concat("The message [", msg.SequentialID, "|", message.ID, "] isn't processable. [", msg.Original, "]"), TraceType.Optional);
                                SendMessageToNormalFlow(message);
                            }
                        }
                        else
                        {
                            _traceWriter.Trace(string.Concat("The message [", msg.SequentialID, "|", message.ID, "] is 0100. [", msg.Original, "]"), TraceType.Optional);
                            SendMessageToNormalFlow(message);
                        }
                    }
                    else
                    {
                        _traceWriter.Trace(string.Concat("The message [", msg.SequentialID, "|", message.ID, "] isn't from MegaBolsa. [", msg.Original, "]"), TraceType.Optional);                        
                        SendMessageToNormalFlow(message);
                    }
                }
            }
            else
            {
                OptimizeMessage(msg);
            }

            _totalRoutedMessagesPerInterval++;
            _totalRoutedMessages++;
        }        

        /// <summary>
        /// 
        /// </summary>
        /// <param name="message"></param>
        private void ProcessMessage0103(MessageEntity.BaseMessage message)
        {
            if (((MessageEntity.STM.Message0103)message).Operator == 0)
            {
                OptimizeMessage(message);
            }
            else
            {
                if (!_dictionaryOrderManager.Exists(message))
                {
                    if(SendMessageToNormalFlow(message))
                        _dictionaryOrderManager.AddOrder(message);
                }
                else
                {
                    SendMessage(message);
                }
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="message"></param>
        private void ProcessMessage0105(MessageEntity.BaseMessage message)
        {
            if (((MessageEntity.STM.Message0105)message).Operator == 0)
            {
                OptimizeMessage(message);
            }
            else
            {
                if (!_dictionaryOrderManager.Exists(message))
                {
                    if(SendMessageToNormalFlow(message))
                        _dictionaryOrderManager.AddOrder(message);
                    
                }
                else
                {
                    SendMessage(message);
                }
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="message"></param>
        private void ProcessMessage0138(MessageEntity.BaseMessage message)
        {
            OptimizeMessage(message);
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="message"></param>
        private void ProcessMessage0172(MessageEntity.BaseMessage message)
        {
            MessageEntity.STM.Message0172 message0172;
            message0172 = (MessageEntity.STM.Message0172)message;

            if (message0172.Operator == 0)
            {
                OptimizeMessage(message);
            }
            else
            {
                if (!IsCancellingOrder(message0172.OrderStatus))
                {
                    if (!_dictionaryOrderManager.Exists(message))
                    {
                        if (SendMessageToNormalFlow(message))
                            _dictionaryOrderManager.AddOrder(message);
                    }
                    else
                    {
                        OptimizeMessage(message);
                    }
                }
                else
                {
                    OptimizeMessage(message);
                }
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="orderStatus"></param>
        /// <returns></returns>
        private bool IsCancellingOrder(char orderStatus)
        {
            return _cancellingStatus.Contains(orderStatus);
        }

        /// <summary>
        /// Sends the message to the appropriated repository.
        /// </summary>
        /// <param name="message"></param>
        private void SendMessage(MessageEntity.BaseMessage message)
        {
            Configuration.Router.DistributionRule repository = null;
            HftProvider hftProvider = null;
            bool sendSucceeded = false;
            bool isRecovery = false;
            repository = GetRepository(message);

            try
            {
                if (_providersList != null)
                {
                    lock (_providersList)
                    {
                        hftProvider = (from item in _providersList
                                       where item.Provider.Path ==
                                        repository.HFTProcess.MessageRepository.ConnectionInformation[
                                            Configuration.Repository.RepositoryPropertiesConstants.ConnectionString]
                                       select item).First();
                    }
                }

                try
                {
                    while (!sendSucceeded)
                    {
                        message.SequentialID = hftProvider.SequentialId;

                        _traceWriter.Trace(string.Concat("SendMessage: the message [", message.SequentialID, "|", message.ID, "] will be send to provider [", hftProvider.Provider.Path, "]"), TraceType.Optional);
                        sendSucceeded = hftProvider.Provider.Send(GetMsmqMessage(message));

                        if (!sendSucceeded)
                        {
                            _traceWriter.Trace(string.Concat("SendMessage: Error when trying to send the message to msmq. ErrorCode: [", hftProvider.Provider.LastErrorCode, "] ErrorMessage: [", hftProvider.Provider.LastErrorMessage, "]"));
                            HandleError(10, string.Concat("SendMessage: Error when trying to send the message to msmq. ErrorCode: [", hftProvider.Provider.LastErrorCode, "] ErrorMessage: [", hftProvider.Provider.LastErrorMessage, "]"), true);
                            isRecovery = true;
                        }
                        else if (sendSucceeded && isRecovery)
                        {
                            HandleRecovery(ref isRecovery);
                        }
                    }

                    _traceWriter.Trace(string.Concat("SendMessage: the message [", message.SequentialID, "|", message.ID, "] has been sent to the provider [", hftProvider.Provider.Path, "]"), TraceType.Optional);

                    hftProvider.SequentialId++;
                    _hftMessages++;
                }
                catch (Exception ex)
                {
                    _traceWriter.Trace(string.Concat("Error when trying to send the message to msmq.", "[", message.Original, "] ", ex.Message, ex.StackTrace));
                    HandleError(1, string.Concat("Error when trying to send the message to msmq.", "[", message.Original, "] ", ex.Message, ex.StackTrace));
                }
            }
            catch (Exception ex)
            {
                _traceWriter.Trace(string.Concat("Error when trying to get hft provider. ", "[", message.Original, "] ", ex.Message, ex.StackTrace));
                HandleError(1, string.Concat("Error when trying to get hft provider. ", "[", message.Original, "] ", ex.Message, ex.StackTrace));
            }
        }

        public void SendResetBroadcast()
        {
            MessageEntity.OMMSystem.OMMSystemMessage oMMSystemMessage = new Sinacor.OMM.MessageEntity.OMMSystem.OMMSystemMessage();
            oMMSystemMessage.OMMSystemMessageType = Sinacor.OMM.MessageEntity.Enum.OMMSystemMessageType.ResetId;

            if (_providersList != null)
            {
                lock (_providersList)
                {
                    foreach (var item in _providersList)
                    {
                        bool sendSucceeded = false;

                        while (!sendSucceeded && !_disposing)
                        {
                            _traceWriter.Trace(string.Concat("SendResetBroadcast: A reset request will be send to the provider ", item.Provider.Path));
                            sendSucceeded = item.Provider.Send(new System.Messaging.Message() { Label = "OMMSystemMessage", Body = oMMSystemMessage });
                            _traceWriter.Trace(string.Concat("SendResetBroadcast: A reset request has been sent to the provider ", item.Provider.Path, " with return ", sendSucceeded, " ErrorCode: [", item.Provider.LastErrorCode, "] ErrorMessage: [", item.Provider.LastErrorMessage, "]"));

                            Sleeper.Sleep(1000, () => _disposing);

                            if (!sendSucceeded)
                            {
                                _traceWriter.Trace(string.Concat("SendResetBroadcast: Error when trying to send reset message to [", item.Provider.Path, "]. Error: ", item.Provider.LastErrorMessage));
                                Sleeper.Sleep(5000, () => _disposing);
                            }
                        }

                        if (_disposing)
                            break;
                    }
                }
            }
        }

        private void ResetProvidersId()
        {
            MessageEntity.OMMSystem.OMMSystemMessage oMMSystemMessage = new Sinacor.OMM.MessageEntity.OMMSystem.OMMSystemMessage();
            oMMSystemMessage.OMMSystemMessageType = Sinacor.OMM.MessageEntity.Enum.OMMSystemMessageType.ResetId;

            try
            {
                if (_providersList != null)
                {
                    lock (_providersList)
                        foreach (var item in _providersList)
                            item.SequentialId = 1;
                }
            }
            catch (Exception ex)
            {
                _traceWriter.Trace(string.Concat("Error when trying to reset the id of the hft providers: ", ex.Message, ex.StackTrace));
                HandleError(1, string.Concat("Error when trying to reset the id of the hft providers: ", ex.Message, ex.StackTrace));
            }
        }

        /// <summary>
        /// Optimizes the message.
        /// </summary>
        /// <param name="message"></param>
        private void OptimizeMessage(MessageEntity.BaseMessage message)
        {
            bool sendSucceeded = false;
            bool hasError = false;

            _traceWriter.Trace(string.Concat("OptimizeMessage: The message [", message.SequentialID, "|", message.ID, "] will be optimized. [", message.Original, "]"), Sinacor.OMM.Util.TraceType.Optional);

            while (!sendSucceeded)
            {
                sendSucceeded = _dumpClient.Send(new DumpEntity.DumpLineData()
                {
                    FolderName = "Roteador",
                    FileName = "OptimizedMessages",
                    DataToDump = string.Concat(message.ID, " ", message.Original)
                });

                if (!sendSucceeded)
                {
                    hasError = true;
                    _traceWriter.Trace(string.Concat("Error when trying to optimize the message [", message.Original, "] Error Code: ", _dumpClient.LastErrorCode, " Error Message: ", _dumpClient.LastErrorMessage));
                    HandleError(1, string.Concat("Error when trying to optimize the message [", message.Original, "] Error Code: ", _dumpClient.LastErrorCode, " Error Message: ", _dumpClient.LastErrorMessage), true);
                    Sleeper.Sleep(500, () => !sendSucceeded);
                }
                else if (sendSucceeded && hasError)
                {
                    HandleRecovery();
                }
            }

            _traceWriter.Trace(string.Concat("OptimizeMessage: The message [", message.SequentialID, "|", message.ID, "] has been optimized. [", message.Original, "]"), Sinacor.OMM.Util.TraceType.Optional);
            _optimizedMessages++;
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="message"></param>
        /// <returns></returns>
        private Configuration.Router.DistributionRule GetRepository(MessageEntity.BaseMessage message)
        {
            Order order = _dictionaryOrderManager.TranslateToOrder(message);
            Configuration.Router.DistributionRule distributionRule = null;
            IEnumerable<Configuration.Router.DistributionRule> repositoryQuery;
            _getRepositorySucceeded = false;

            while (!_getRepositorySucceeded)
            {
                repositoryQuery = from item in RouterConfigurationManager.RouterConfiguration.DistributionRules
                                  where item.ClientId == order.Client &&
                                  (order.Operator >= item.StartPort && order.Operator <= item.EndPort)
                                  select item;

                if (repositoryQuery.Count() > 0)
                {
                    distributionRule = repositoryQuery.ToList()[0];
                    _getRepositorySucceeded = true;
                }
                else
                {
                    repositoryQuery = from item in RouterConfigurationManager.RouterConfiguration.DistributionRules
                                      where item.ClientId == null &&
                                      (order.Operator >= item.StartPort && order.Operator <= item.EndPort)
                                      select item;

                    if (repositoryQuery.Count() > 0)
                    {
                        distributionRule = repositoryQuery.ToList()[0];
                        _getRepositorySucceeded = true;
                    }
                    else
                    {
                        _traceWriter.Trace(string.Concat("Error when trying to get default hft provider [", message.Original, "]"));
                        Sleeper.Sleep(1000, () => !_getRepositorySucceeded);
                    }
                }
            }

            return distributionRule;
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="message"></param>
        /// <returns></returns>
        private bool IsProcessable(MessageEntity.BaseMessage message)
        {
            return (
                message is MessageEntity.STM.Message0103 ||
                message is MessageEntity.STM.Message0105 ||
                message is MessageEntity.STM.Message0172 ||
                message is MessageEntity.STM.Message0138);
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="message"></param>
        /// <returns></returns>
        private bool IsHFT(MessageEntity.BaseMessage message)
        {
            Order order = null;
            order = _dictionaryOrderManager.TranslateToOrder(message);
            IEnumerable<Configuration.Router.DistributionRule> query;

            query = from o in RouterConfigurationManager.RouterConfiguration.DistributionRules
                    where
                    (o.ClientId == order.Client && (order.Operator >= o.StartPort && order.Operator <= o.EndPort)) ||
                    (o.ClientId == null && (order.Operator >= o.StartPort && order.Operator <= o.EndPort))
                    select o;

            if (query != null && query.Count() > 0)
            {
                return true;
            }
            else
            {
                return false;
            }
        }

        /// <summary>
        /// 
        /// </summary>
        private void LoadProviders()
        {
            _providersList = new List<HftProvider>();
            MsmqSender msmqSender = null;
            List<Configuration.Router.DistributionRule> rules = RouterConfigurationManager.RouterConfiguration.DistributionRules;

            try
            {
                lock (rules)
                {
                    for (int i = 0; i < rules.Count; i++)
                    {
                        if (_providersList.Find(p => p.Provider.Path ==
                                rules[i].HFTProcess.MessageRepository.ConnectionInformation[Configuration.Repository.RepositoryPropertiesConstants.ConnectionString]) == null)
                        {
                            msmqSender = _msmqHelper.CreateRepository(rules[i].HFTProcess.MessageRepository);
                            msmqSender.OnHealthStatusChanged += new MsmqSender.HealthStatusChanged(msmqSender_OnHealthStatusChanged);
                            _providersList.Add(new HftProvider() { Provider = msmqSender, SequentialId = 1 });
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                _traceWriter.Trace(string.Concat("Error when loading hft providers: ", ex.Message, ex.StackTrace));
                HandleError(1, string.Concat("Error when loading hft providers: ", ex.Message, ex.StackTrace));
            }
        }

        void msmqSender_OnHealthStatusChanged(MsmqSender MsmqSender)
        {
            bool queueExceedsLimit = false;

            if (!_queueStatusChanged.ContainsKey(MsmqSender.Path))
            {
                _queueStatusChanged.Add(MsmqSender.Path, MsmqSender.QueueExceedsLimit);

                if (_queueStatusChanged.Count == _providersList.Count)
                {
                    foreach (var item in _queueStatusChanged)
                    {
                        if (item.Value)
                        {
                            queueExceedsLimit = true;
                            break;
                        }
                    }

                    _queueStatusChanged.Clear();
                }
            }

            if (queueExceedsLimit)
            {
                if (OnError != null)
                    OnError(this, ProcessEntity.RouterEventArgs.Empty);
            }
            else if (!queueExceedsLimit && _routerShdms.RouterDataInfo.StopedBySomeQueueExceedsLimit)
            {
                if (OnRecovery != null)
                    OnRecovery(this, ProcessEntity.RouterEventArgs.Empty);
            }

            _routerShdms.RouterDataInfo.StopedBySomeQueueExceedsLimit = queueExceedsLimit;
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="messages"></param>
        /// <param name="queueId"></param>                
        private void DequeueNormalFlowMessage(PSE.Framework.Messaging.SynchronizedQueue<MessageEntity.BaseMessage> sender, List<MessageEntity.BaseMessage> messages, string queueId)
        {
            List<MessageEntity.STM.BaseMessage> megaBolsaMessages = new List<MessageEntity.STM.BaseMessage>();
            List<MessageEntity.STM.BaseMessage> cblcMessages = new List<MessageEntity.STM.BaseMessage>();
            bool insertSucceeded = false;
            bool isRecovery = false;

            try
            {
                // creates the lists of MegaBolsa messages and CBLC messages
                foreach (MessageEntity.BaseMessage item in messages)
                {
                    if (((MessageEntity.STM.BaseMessage)item).Source == Sinacor.OMM.MessageEntity.Enum.MessageSource.CBLC)
                    {
                        //cblcMessages.Add(new QueueOrder() { OperationDate = item.OperationDate, System = GetSystem(item), Sequential = item.SequentialID, Signal = item.Original, Id = item.ID });
                        ((MessageEntity.STM.BaseMessage)item).System = GetSystem(item);
                        cblcMessages.Add((MessageEntity.STM.BaseMessage)item);
                    }
                    else
                    {
                        //megaBolsaMessages.Add(new QueueOrder() { OperationDate = item.OperationDate, Sequential = item.SequentialID, Signal = item.Original, Id = item.ID });
                        megaBolsaMessages.Add((MessageEntity.STM.BaseMessage)item);
                    }
                }

                InsertMessages(cblcMessages, ref insertSucceeded, ref isRecovery, MessageEntity.Enum.MessageSource.CBLC);
                InsertMessages(megaBolsaMessages, ref insertSucceeded, ref isRecovery, MessageEntity.Enum.MessageSource.MegaBolsa);

                _bulkInsertSize += cblcMessages.Count + megaBolsaMessages.Count;
                _bulkInsertTimes++;
            }
            catch (Exception ex)
            {
                _traceWriter.Trace(string.Concat("Error when trying to dequeue the normal flow messages: ", ex.Message, ex.StackTrace));
                HandleError(1, string.Concat("Error when trying to dequeue the normal flow messages: ", ex.Message, ex.StackTrace));
            }
            finally
            {
                cblcMessages.Clear();
                megaBolsaMessages.Clear();
            }
        }

        private string GetSystem(MessageEntity.BaseMessage message)
        {
            string system = string.Empty;

            if (message is MessageEntity.STM.MessageAN)
                system = ((MessageEntity.STM.MessageAN)message).System;

            if (message is MessageEntity.STM.MessageAN53)
                system = ((MessageEntity.STM.MessageAN53)message).System;

            if (message is MessageEntity.STM.MessageANF)
                system = ((MessageEntity.STM.MessageANF)message).System;

            return system;
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="messages"></param>
        /// <param name="insertSucceeded"></param>
        /// <param name="isRecovery"></param>
        private void InsertMessages(List<MessageEntity.STM.BaseMessage> messages, ref bool insertSucceeded, ref bool isRecovery, MessageEntity.Enum.MessageSource source)
        {
            Stopwatch stopwatch = new Stopwatch();

            if (messages.Count > 0)
            {
                insertSucceeded = false;
                // attempts to do the insert until it succeeds.
                while (!insertSucceeded)
                {
                    try
                    {
                        // sends the orders to the repository
                        _traceWriter.Trace(string.Concat("InsertMessages: the messages will be inserted to database."), Sinacor.OMM.Util.TraceType.Optional);

                        stopwatch.Start();

                        if (source == Sinacor.OMM.MessageEntity.Enum.MessageSource.CBLC)
                        {
                            _orderResourceAccess.InsertCblcOrders(messages);
                        }
                        else
                        {
                            _orderResourceAccess.InsertMegaBolsaOrders(messages);
                        }

                        stopwatch.Stop();

                        _traceWriter.Trace(string.Concat("InsertMessages: the messages have been inserted to database."), Sinacor.OMM.Util.TraceType.Optional);

                        _insertOperationTime += stopwatch.Elapsed.Seconds;
                        insertSucceeded = true;
                        HandleRecovery(ref isRecovery);
                    }
                    catch (System.Data.Common.DbException ex)
                    {
                        _traceWriter.Trace(string.Concat("Error when trying to insert the messages to database, DbException: ", ex.Message, ex.StackTrace));
                        insertSucceeded = false;
                        isRecovery = true;
                        HandleError(10000, string.Concat("Error when trying to insert the messages to database, DbException: ", ex.Message, ex.StackTrace), true);
                    }
                    catch (Exception ex)
                    {
                        _traceWriter.Trace(string.Concat("Error when trying to insert the messages to database, GeneralException: ", ex.Message, ex.StackTrace));
                        insertSucceeded = false;
                        isRecovery = true;
                        HandleError(10, string.Concat("Error when trying to insert the messages to database, GeneralException: ", ex.Message, ex.StackTrace), true);
                    }
                }
            }
        }

        /// <summary>
        /// 
        /// </summary>
        private void HandleError(int miliseconds, string message, bool shouldPause)
        {
            if (OnError != null)
                OnError(this, new ProcessEntity.RouterEventArgs() { ErrorMessage = message, ShouldPause = shouldPause });

            Sleeper.Sleep(miliseconds, IsShouldStopProcessing);
        }

        private void HandleError(int miliseconds, string message)
        {
            if (OnError != null)
                OnError(this, new ProcessEntity.RouterEventArgs() { ErrorMessage = message });

            Sleeper.Sleep(miliseconds, IsShouldStopProcessing);
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="isRecovery"></param>
        private void HandleRecovery(ref bool isRecovery)
        {
            if (isRecovery)
            {
                isRecovery = false;

                if (OnRecovery != null)
                    OnRecovery(this, ProcessEntity.RouterEventArgs.Empty);
            }
        }

        private void HandleRecovery()
        {
            if (OnRecovery != null)
                OnRecovery(this, ProcessEntity.RouterEventArgs.Empty);
        }

        /// <summary>
        /// 
        /// </summary>
        /// <returns></returns>
        private bool IsShouldStopProcessing()
        {
            return !_shouldStopProcessing;
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="message"></param>
        ///         
        public bool SendMessageToNormalFlow(Sinacor.OMM.MessageEntity.BaseMessage message)
        {
            bool _sendMessageToNormalFlow = false;
            bool isMessageIdGreater = true;
            MessageEntity.STM.BaseMessage stmBaseMessage = (MessageEntity.STM.BaseMessage)message;

            try
            {
                if (stmBaseMessage.Source == Sinacor.OMM.MessageEntity.Enum.MessageSource.MegaBolsa)
                {
                    isMessageIdGreater = Util.MessageHelper.IsMessageIdGreaterThanLast(message.ID, _lastMegaBolsaMessageId);

                    if (!isMessageIdGreater)
                        _traceWriterOtimizadas.Trace("SendMessageToNormalFlow: The message [" + message.SequentialID + "|" + message.ID + " LastMegaBolsaMessageId: " + _lastMegaBolsaMessageId + " ] has already been enqueued to the normal flow. [" + message.Original + "]");
                }
                else
                {
                    isMessageIdGreater = Util.MessageHelper.IsMessageIdGreaterThanLast(message.ID, _lastCblcMessageId);

                    if(!isMessageIdGreater)
                        _traceWriterOtimizadas.Trace("SendMessageToNormalFlow: The message [" + message.SequentialID + "|" + message.ID + " LastCblcMessageId: " + _lastCblcMessageId + " ] has already been enqueued to the normal flow. [" + message.Original + "]");

                }

                if (isMessageIdGreater)
                {
                    _traceWriter.Trace(string.Concat("SendMessageToNormalFlow: The message [", message.SequentialID, "|", message.ID, "] will be enqueued to the normal flow. [", message.Original, "]"), Sinacor.OMM.Util.TraceType.Optional);
                    _internalQueueManager.Enqueue(ROUTER_MESSAGE_PROCESSOR_QUEUE_ID, message);
                    _traceWriter.Trace(string.Concat("SendMessageToNormalFlow: The message [", message.SequentialID, "|", message.ID, "] has been enqueued to the normal flow. [", message.Original, "]"), Sinacor.OMM.Util.TraceType.Optional);

                    _routedMessagesToNormalFlow++;

                    if (stmBaseMessage.Source == Sinacor.OMM.MessageEntity.Enum.MessageSource.MegaBolsa)
                    {
                        _lastMegaBolsaMessageId = message.ID;
                    }
                    else
                    {
                        _lastCblcMessageId = message.ID;
                    }
                    _sendMessageToNormalFlow = true;
                }
                else
                {
                    _traceWriter.Trace(string.Concat("SendMessageToNormalFlow: The message [", message.SequentialID, "|", message.ID, "] has already been enqueued to the normal flow. [", message.Original, "]"), Sinacor.OMM.Util.TraceType.Optional);
                }
            }
            catch (Exception ex)
            {
                _traceWriter.Trace(string.Concat("Error when trying to send the message to normal flow queue: ", ex.Message, ex.StackTrace));
                HandleError(1, string.Concat("Error when trying to send the message to normal flow queue: ", ex.Message, ex.StackTrace));
            }

            return _sendMessageToNormalFlow;
        }        

        /// <summary>
        /// 
        /// </summary>
        /// <param name="message"></param>
        /// <returns></returns>
        private System.Messaging.Message GetMsmqMessage(MessageEntity.BaseMessage message)
        {
            char separator = '|';
            return new System.Messaging.Message() { Label = string.Concat(message.SequentialID, separator, message.ID), Body = message.Original };
        }

        /// <summary>
        /// 
        /// </summary>
        /// <returns></returns>
        public string GetLastProcessedMessageId()
        {
            List<string> messageIds = new List<string>();
            string minorMessageId = string.Empty;

            try
            {
                lock (_orderResourceAccess)
                    messageIds = _orderResourceAccess.GetLastProcessedMessageIds(RouterConfigurationManager.OrderBovespaSystemInformation.SystemDate);

                if (messageIds.Count > 0)
                    minorMessageId = GetLastProcessedMessageId(messageIds);
            }
            catch (Exception ex)
            {
                _traceWriter.Trace(string.Concat("Error when trying to get the last processed message id: ", ex.Message, ex.StackTrace));
                HandleError(1, string.Concat("Error when trying to get the last processed message id: ", ex.Message, ex.StackTrace));
            }

            return minorMessageId;
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="messageIds"></param>
        /// <returns></returns>
        private string GetLastProcessedMessageId(List<string> messageIds)
        {
            string minorMessageId = string.Empty;

            try
            {
                if (messageIds.Count > 0)
                {
                    byte minorSequence = (from m in messageIds
                                          orderby
                                          Convert.ToByte(m.Substring(8, 1)[0])
                                          select Convert.ToByte(m.Substring(8, 1)[0])).First();

                    minorMessageId = (from m in messageIds
                                      where Convert.ToByte(m.Substring(8, 1)[0]) == minorSequence
                                      orderby Convert.ToInt32(m.Substring(9, 10))
                                      select m).First();
                }
            }
            catch (Exception ex)
            {
                _traceWriter.Trace(string.Concat("Error when trying to get the last processed message id: ", ex.Message, ex.StackTrace));
                HandleError(1, string.Concat("Error when trying to get the last processed message id: ", ex.Message, ex.StackTrace));
            }

            return minorMessageId;
        }

        private void SetLastMegaBolsaMessageId()
        {
            try
            {
                lock (_orderResourceAccess)
                    _lastMegaBolsaMessageId = _orderResourceAccess.GetLastMegaBolsaMessageId(RouterConfigurationManager.OrderBovespaSystemInformation.SystemDate);
            }
            catch (Exception ex)
            {
                _traceWriter.Trace(string.Concat("Error when trying to get the last mega bolsa processed message id: ", ex.Message, ex.StackTrace));
                HandleError(1, string.Concat("Error when trying to get the last mega bolsa processed message id: ", ex.Message, ex.StackTrace));
                throw ex;
            }
        }

        private void SetLastCblcMessageId()
        {
            try
            {
                lock (_orderResourceAccess)
                    _lastCblcMessageId = _orderResourceAccess.GetLastCblcMessageId(RouterConfigurationManager.OrderBovespaSystemInformation.SystemDate);
            }
            catch (Exception ex)
            {
                _traceWriter.Trace(string.Concat("Error when trying to get the last cblc processed message id: ", ex.Message, ex.StackTrace));
                HandleError(1, string.Concat("Error when trying to get the last cblc processed message id: ", ex.Message, ex.StackTrace));
                throw ex;
            }
        }


        /// <summary>
        /// Recycles the configuration of the message processor.
        /// </summary>
        public void Recycle()
        {
            _lastProcessedMessageId = string.Empty;
            _routedMessagesToNormalFlow = 0;
            _optimizedMessages = 0;
            _hftMessages = 0;
            _totalRoutedMessagesPerInterval = 0;
            _totalRoutedMessages = 0;
            _bulkInsertSize = 0;
            _bulkInsertTimes = 0;
            _insertOperationTime = 0;
            _lastHealthStatusChecked = new DateTime(1, 1, 1);
            _lastHealthDBStatusChecked = new DateTime(1, 1, 1);
            _lastMegaBolsaMessageId = string.Empty;
            _lastCblcMessageId = string.Empty;

            ResetProvidersId();
            SendResetBroadcast();
            _dictionaryOrderManager.LoadOrders();
            _orderResourceAccess.ResetSequence();
            _orderResourceAccess.ResetSequentialNumber();
        }

        #region IDisposable Members

        public void Dispose()
        {
            _disposing = true;
            _shouldStopCollecting = true;
            _shouldStopCheckingDatabaseQueue = true;
            _shouldStopProcessing = true;
            _getRepositorySucceeded = true;

            if (_orderResourceAccess != null)
                _orderResourceAccess.Dispose();

            _orderResourceAccess = null;
            _msmqHelper = null;

            if (_providersList != null)
                for (int i = 0; i < _providersList.Count; i++)
                    _providersList[i].Provider.Dispose();

            _providersList = null;

            if (_systemMessageProvider != null)
                _systemMessageProvider.Dispose();
            _systemMessageProvider = null;
            if (_internalQueueManager != null)
                _internalQueueManager.Dispose();
            _internalQueueManager = null;

            if (_dictionaryOrderManager != null)
                _dictionaryOrderManager.Dispose();
            _dictionaryOrderManager = null;

            if (_dataBaseCommunicator != null)
                _dataBaseCommunicator.Dispose();
            _dataBaseCommunicator = null;

            if (_dumpClient != null)
                _dumpClient.Dispose();
            _dumpClient = null;

            _traceWriter = null;
        }

        #endregion

        public void CheckRoutedMessagesCount()
        {
            if (_orderResourceAccess.GetRoutedMessagesCount(RouterConfigurationManager.OrderBovespaSystemInformation.SystemDate) == 0)
            {
                _orderResourceAccess.ResetSequence();
                _orderResourceAccess.ResetSequentialNumber();
            }
        }
    }

    public class HftProvider
    {
        public MsmqSender Provider { get; set; }
        public long SequentialId { get; set; }
    }
}
